package cn.juque.lucenecandy.threads;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.ClassUtil;
import cn.juque.common.base.BaseResponseVO;
import cn.juque.common.base.PageInfo;
import cn.juque.common.constants.MessageEnum;
import cn.juque.common.fastjson.abstracts.AbstractTypeReference;
import cn.juque.lucenecandy.bo.IndexBO;
import cn.juque.lucenecandy.cache.IndexInfoCache;
import cn.juque.lucenecandy.cache.IpCache;
import cn.juque.lucenecandy.core.base.BaseMessageEntity;
import cn.juque.lucenecandy.core.constants.StrConstant;
import cn.juque.lucenecandy.core.constants.UrlConstant;
import cn.juque.lucenecandy.core.enums.DataSyncMsgTypeEnum;
import cn.juque.lucenecandy.core.utils.SignatureUtil;
import cn.juque.lucenecandy.helper.DocumentHelper;
import cn.juque.lucenecandy.helper.LuceneHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.IntPoint;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author juque
 * @version 1.0.0
 * <ul>
 *     <li>DataSyncMsgReadThreaPool</li>
 * </ul>
 * @date 2024-01-03
 **/
@Slf4j
@Component("dataSyncMsgReadThreadPool")
public class DataSyncMsgReadThreadPool {

    private static final ThreadFactory THREAD_FACTORY = ThreadUtil.newNamedThreadFactory("index-msg-read-", true);

    private static final ScheduledThreadPoolExecutor EXECUTOR = new ScheduledThreadPoolExecutor(1, THREAD_FACTORY);

    private static final BooleanQuery.Builder BOOL_BUILDER = new BooleanQuery.Builder();

    private static final PageInfo PAGE_INFO = new PageInfo();

    private static final Sort SORT = new Sort(new SortField("timestamp", SortField.Type.LONG));

    private static final AtomicBoolean IS_END = new AtomicBoolean(true);

    @Resource
    private DocumentHelper documentHelper;

    @Resource
    private LuceneHelper luceneHelper;

    @Resource
    private IpCache ipCache;

    @Resource
    private IndexInfoCache indexInfoCache;

    public void init() {
        BOOL_BUILDER.add(IntPoint.newExactQuery(
                "msg_type", DataSyncMsgTypeEnum.WAIT_SEND.getCode()), BooleanClause.Occur.MUST);
        PAGE_INFO.setPage(1);
        PAGE_INFO.setLimit(100);
        EXECUTOR.scheduleWithFixedDelay(this::execute, 0,5, TimeUnit.SECONDS);
    }

    private void execute() {
        if (!IS_END.compareAndSet(true, false)) {
            return;
        }
        String className = ClassUtil.getClassName(BaseMessageEntity.class, false);
        IndexBO indexBO = this.indexInfoCache.get(className, false);
        String indexName = indexBO.getIndexName();
        try {
            List<Document> docList = this.luceneHelper.searchByPage(indexName, BOOL_BUILDER, SORT, PAGE_INFO, null);
            while (CollUtil.isNotEmpty(docList)) {
                List<BaseMessageEntity> entityList = this.documentHelper.toEntityList(docList, BaseMessageEntity.class);
                this.batchPost(entityList);
                // 删除已同步的文档
                BooleanQuery.Builder builder = new BooleanQuery.Builder();
                entityList.forEach(
                        f -> builder.add(new TermQuery(new Term(StrConstant.D_ID, f.getId())), BooleanClause.Occur.SHOULD));
                this.luceneHelper.deleteDocuments(indexName, builder.build());
                docList = this.luceneHelper.searchByPage(indexName, BOOL_BUILDER, SORT, PAGE_INFO, null);
            }
        } catch (Exception e) {
            log.error("文档同步异常", e);
        } finally {
            IS_END.set(true);
        }
    }

    private void batchPost(List<BaseMessageEntity> entityList) {
        List<String> ips = this.ipCache.listValidIp(false);
        try {
            Map<String, BaseResponseVO<Boolean>> result = this.ipCache.batchPost(ips,
                    SignatureUtil.buildUrl(UrlConstant.BATCH_MSG),
                    new AbstractTypeReference<BaseResponseVO<Boolean>>() {}, entityList);
            result.forEach((k,v)->{
                if(MessageEnum.OK.getCode().equals(v.getCode()) && Boolean.TRUE.equals(v.getData())) {
                    return;
                }
                log.error("ip:{} sync msg-data by post error, errorCode={}, errorMsg={}", k, v.getCode(), v.getMsg());
            });
            boolean flag = result.values().stream().anyMatch(f -> !Boolean.TRUE.equals(f.getData()));
            if (flag) {
                log.warn("文档同步存在失败");
            }
        } catch (Exception e) {
            log.error("sync message error", e);
        }
    }
}
